package com.flute.haflute.agent.topo;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.heartbeat.HeartBeatState;
import com.flute.haflute.agent.heartbeat.HeartBeatState.STATUS;
import com.flute.haflute.agent.heartbeat.NodeStateUpdateResult;
import com.flute.haflute.alarm.AlarmService;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterConstants.ROLE;
import com.flute.haflute.net.messaging.Message;
import com.flute.haflute.net.messaging.Message.MessageType;
import com.flute.haflute.net.messaging.MessageFactory;

/**
 * 集群拓扑数据库<br>
 * 集群拓扑相关的维护和数据库状态都在这里
 * 
 */
public class ClusterTopoDBService {

	private static Logger logger = LoggerFactory
			.getLogger(ClusterTopoDBService.class);

	private static ClusterTopoDBService instance = new ClusterTopoDBService();

	private Node dM;
	private Node bDM;

	/** nodeRoleDB 存储<节点，角色>对的数据库 */
	private Map<Node, ROLE> nodeRoleDB;

	/** neighborStateDB 存储<节点，状态>对的数据库 */
	private Map<Node, HeartBeatState> neighborStateDB;

	private ClusterTopoDBService() {
		nodeRoleDB = new ConcurrentHashMap<Node, ROLE>(10);
		neighborStateDB = new ConcurrentHashMap<Node, HeartBeatState>(10);
	}

	public static ClusterTopoDBService getInstance() {
		return instance;
	}

	/**
	 * 仅更新角色状态，agent使用，其他模块请不要使用
	 * 
	 * @param node
	 * @param role
	 * @return 老的
	 */
	public ROLE updateNodeRole(Node node, ROLE role) {
		return nodeRoleDB.put(node, role);
	}

	/**
	 * 根据心跳消息更新节点状态，
	 * 
	 * @param heartBeatMessage
	 * @return 更新的结果
	 */
	public NodeStateUpdateResult updateNodeState(Message heartBeatMessage) {
		
		NodeStateUpdateResult result = new NodeStateUpdateResult();
		Node node = new Node(heartBeatMessage.getFrom().getHostAddress());
		HeartBeatState state = neighborStateDB.get(node);
		boolean isNew = false;
		if (state == null) {
			state = new HeartBeatState();
			isNew = true;
			logger.info("new node ({}) connected", node.getIp());
		}

		if (heartBeatMessage.getMessageType().equals(MessageType.HB.name())) {
			// 本轮接收到第一阶段心跳消息
			state.setStatus(STATUS.ONEWAY);
			String roleString = new String(heartBeatMessage.getHeader("role"));
			ROLE newRole = ROLE.valueOf(roleString);
			ROLE oldRole = nodeRoleDB.get(node);
			if (newRole != oldRole) {
				if (isNew) {
					addNewNode(node, newRole, state);
				}
			}
			result.setFinished(false);
			result.setMessage(MessageFactory.getHeartBeatAckMessage());
		} else if (heartBeatMessage.getMessageType().equals(
				MessageType.HBACK.name())) {
			// 本轮接收到第二阶段心跳消息
			state.setStatus(STATUS.ACTIVE);
			state.setActiveTime(now());
			result.setFinished(true);
		} else if (heartBeatMessage.getMessageType().equals(
				MessageType.HBACK2.name())) {
			// 本轮接收到第三阶段心跳消息
			state.setStatus(STATUS.ACTIVE);
			state.setActiveTime(now());
			result.setFinished(true);
		}

		return result;
	}

	public void setDM(Node dm) {
		dM = dm;
		nodeRoleDB.put(dm, ClusterConstants.ROLE.DM);
	}

	public Node getDM() {
		return dM;
	}

	public Node getBDM() {
		return bDM;
	}

	public void setBDM(Node bdm) {
		bDM = bdm;
		nodeRoleDB.put(bdm, ROLE.BDM);
	}

	public Map<Node, ROLE> getNodeRoleDB() {
		return nodeRoleDB;
	}

	public Map<Node, HeartBeatState> getNeighborStateDB() {
		return neighborStateDB;
	}

	/**
	 * 从拓扑数据库中清除一个节点
	 */
	public synchronized void removeNode(Node node) {
		neighborStateDB.remove(node);
		nodeRoleDB.remove(node);
		logger.info("one node removed: {}", node);
		AlarmService.getInstance().reportLostNode(node);
	}

	/**
	 * 拓扑新发现一个节点
	 */
	private synchronized void addNewNode(Node node, ROLE newRole,
			HeartBeatState state) {
		nodeRoleDB.put(node, newRole);
		neighborStateDB.put(node, state);
		AlarmService.getInstance().reportAddNewNode(node);
	}

	private long now() {
		return System.currentTimeMillis();
	}
}